package org.example.topic;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

@Slf4j
public class TopicSubscriber {
    public static final String USER_NAME = ActiveMQConnection.DEFAULT_USER;

    public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;

    public static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;


    public static void createTopicSubscriber() throws JMSException, InterruptedException {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USER_NAME, PASSWORD, BROKER_URL);
        Connection connection = factory.createConnection();
        connection.start();
        // 不需要事务
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        Topic topic = session.createTopic("activeMQ-topic");

        MessageConsumer consumer = session.createConsumer(topic);

        consumer.setMessageListener(new MessageListener() {
            @SneakyThrows
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                log.info("consumer Message is ：{}", text);
            }
        });

        MessageConsumer consumer2 = session.createConsumer(topic);
        consumer2.setMessageListener(new MessageListener() {
            @SneakyThrows
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                log.info("consumer2 Message is ：{}", text);
            }
        });

        MessageConsumer consumer3 = session.createConsumer(topic);
        consumer3.setMessageListener(new MessageListener() {
            @SneakyThrows
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                log.info("consumer3 Message is ：{}", text);
            }
        });

        Thread.sleep(100 * 1000);
        session.close();
        connection.close();

    }
}
